package com.wang.proxy;

import com.alibaba.fastjson.JSON;
import com.lagou.rpc.common.RpcRequest;
import com.lagou.rpc.common.RpcResponse;
import com.lagou.rpc.config.ZkConfig;
import com.lagou.rpc.listener.NodeChangeListener;
import com.lagou.rpc.registry.ZookeeperRegistry;
import com.wang.client.RpcClient;
import com.wang.port.PollingPort;
import com.wang.port.RpcPort;
import com.wang.strategy.LoadBalanceStrategy;
import com.wang.strategy.MultiLoadBalanceStrategy;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.springframework.util.CollectionUtils;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;

/**
 * 客户端代理类 创建代理对象
 * 1、封装request请求对象
 * 2、创建rpcClient对象
 * 3、发送消息
 * 4、返回结果
 * Created by wangq on 2021/8/29.
 */
public class RpcClientProxy implements NodeChangeListener {
    private static List<RpcPort> serverList = PollingPort.getInstance().getServerList();

    private static int rpcIndex = 0;

    private static LoadBalanceStrategy strategy;
    private static ZookeeperRegistry zookeeperRegistry;
    private static Map<String, RpcClient> rpcClientMap;

    public RpcClientProxy() throws Exception {
        this.zookeeperRegistry = new ZookeeperRegistry();
        this.strategy = new MultiLoadBalanceStrategy();
        this.rpcClientMap = new HashMap<>();

        List<String> discovery = zookeeperRegistry.discovery();
        if (!CollectionUtils.isEmpty(discovery)) {
            discovery.forEach(addr -> {
                String[] split = addr.split(ZkConfig.COLON);
                RpcClient rpcClient = new RpcClient(split[0], Integer.parseInt(split[1]));
                rpcClientMap.put(addr, rpcClient);
            });
        }
        zookeeperRegistry.addListener(this);
    }

    public static Object createProxy(Class serviceClass) {
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class[]{serviceClass}, new InvocationHandler() {
            @Override
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                // 1、封装request请求对象
                RpcRequest rpcRequest = new RpcRequest();
                rpcRequest.setRequestId(UUID.randomUUID().toString());
                rpcRequest.setClassName(method.getDeclaringClass().getName());
                rpcRequest.setMethodName(method.getName());
                rpcRequest.setParameterTypes(method.getParameterTypes());
                rpcRequest.setParameters(args);

                //2、创建rpcClient对象
//                RpcPort rpcPort = getPollingPort();
//                RpcClient rpcClient = new RpcClient(rpcPort.getHost(), rpcPort.getPort());

                RpcClient rpcClient = strategy.getClient(rpcClientMap);
                try {
                    // 3、发送消息
                    Object responseMsgJson = rpcClient.send(JSON.toJSONString(rpcRequest));
                    RpcResponse rpcResponse = JSON.parseObject(responseMsgJson.toString(), RpcResponse.class);
                    if (rpcResponse.getError() != null) {
                        throw new RuntimeException(rpcResponse.getError());
                    }
                    // 4、返回结果
                    Object result = rpcResponse.getResult();
                    Object object = JSON.parseObject(result.toString(), method.getReturnType());
                    return object;
                } catch (Exception exception) {
                    throw exception;
                } finally {
                    rpcClient.close();
                }
            }
        });
    }

    private static RpcPort getPollingPort() {
        System.out.println("当前轮询rpcIndex=" + rpcIndex);
        RpcPort rpcPort = serverList.get(rpcIndex);
        rpcIndex += 1;
        if (rpcIndex >= serverList.size()) {
            rpcIndex = 0;
            System.out.println("轮询完成，重置rpcIndex======" + rpcIndex);
        }
        return rpcPort;
    }


    @Override
    public void update(PathChildrenCacheEvent event) {
        System.out.println(event.getType());
        String path = event.getData().getPath();
        path = path.substring(path.lastIndexOf(ZkConfig.SLASH) + 1);
        switch (event.getType()) {
            case CHILD_ADDED:
            case CONNECTION_RECONNECTED:
            case INITIALIZED:
                String[] split = path.split(ZkConfig.COLON);
                RpcClient rpcClient = new RpcClient(split[0], Integer.parseInt(split[1]));
                rpcClientMap.put(path, rpcClient);
                System.out.println("新增节点:" + path);
                break;
            case CHILD_REMOVED:
            case CONNECTION_LOST:
            case CONNECTION_SUSPENDED:
                rpcClientMap.remove(path);
                System.out.println("移除节点:" + path);
                break;
            default:
                break;
        }
    }
}
